package com.k.activemq;

import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.command.ActiveMQBytesMessage;

import javax.jms.*;
import java.nio.charset.StandardCharsets;

/**
 * @author k
 */
public class ActiveMQUtils {
    public static MqConnectionEntity getConnectionData() {
        MqConnectionEntity connectionEntity = MqConnectionEntity.newBuilder()
                .setUrl("tcp://localhost:61616")
                .setUsername("admin")
                .setPassword("admin")
                .setDestination("ljw-topic")
                //.setDestinationType("queue")
                .setPubSubDomain(true)
                .setMessage("hello")
                .setByteMessage(true)
                .build();
        return connectionEntity;
    }

    public static ActiveMQConnectionFactory getActiveMQConnectionFactory(String url) {
        return new ActiveMQConnectionFactory(url);
    }

    public static String byteMessageToString(ActiveMQBytesMessage message) throws JMSException {
        int len = (int) message.getBodyLength();
        byte[] bytes = new byte[len];
        message.readBytes(bytes);
        return new String(bytes);
    }

    public static String byteMessageToString(BytesMessage message) throws JMSException {
        int len = (int) message.getBodyLength();
        byte[] bytes = new byte[len];
        message.readBytes(bytes);
        return new String(bytes);
    }


    public static String formatMsg(String type, String name, String msg) {
        return String.format("%s - %s - %s", type, name, msg);
    }

    public static void pushToMq(MqConnectionEntity et) throws JMSException {
        // 1.connectionFactory
        ConnectionFactory factory = new ActiveMQConnectionFactory(et.getUrl());

        // 2. connection
        Connection connection = factory.createConnection(et.getUsername(), et.getPassword());
        connection.start();

        // 3. session
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

        // 4. destination
        Destination dest = (et.getPubSubDomain() == true) ? session.createTopic(et.getDestination()) : session.createQueue(et.getDestination());

        // 5. producer
        MessageProducer producer = session.createProducer(dest);
        producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
        if (et.getByteMessage()) {
            System.out.println("send byteMsg:" + et.getMessage());
            byte[] byteMsg = et.getMessage().getBytes(StandardCharsets.UTF_8);
            BytesMessage msg = session.createBytesMessage();
            msg.writeBytes(byteMsg);
            producer.send(msg);
            System.out.println("send byteMsg Ok");
        } else {
            for (int i = 1; i <= 100; i++) {
                TextMessage msg = session.createTextMessage(i + "-hello");
                msg.setIntProperty("id", i);
                producer.send(msg);
                if ((i % 100) == 0) {
                    System.out.println(String.format("Sent %d messages", i));
                }
            }
        }
        producer.send(session.createTextMessage("SHUTDOWN"));
        connection.close();
    }

    public static void listener(MqConnectionEntity connEntity, String listenerName) throws JMSException {
        // 创建连接工厂
        ConnectionFactory connectionFactory = ActiveMQUtils.getActiveMQConnectionFactory(connEntity.getUrl());
        // 创建连接
        Connection connection = connectionFactory.createConnection();
        // 打开连接
        connection.start();
        // 创建会话
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        // 创建队列目标,并标识队列名称，消费者根据队列名称接收数据
        Destination destination = connEntity.getPubSubDomain() ? session.createTopic(connEntity.getDestination()) : session.createQueue(connEntity.getDestination());

        session.createTopic(connEntity.getDestination());
        // 创建消费者
        MessageConsumer consumer = session.createConsumer(destination);
        // 创建消费的监听
        consumer.setMessageListener(new MessageListener() {
            @Override
            public void onMessage(Message message) {
                if (message instanceof TextMessage) {
                    TextMessage textMessage = (TextMessage) message;
                    try {
                        String msgStr = textMessage.getText();
                        System.out.println(ActiveMQUtils.formatMsg("接收消息：" + "TextMessage", Thread.currentThread() + "-" + listenerName, msgStr));
                    } catch (JMSException e) {
                        e.printStackTrace();
                    }
                } else if (message instanceof BytesMessage) {
                    BytesMessage byteMessage = (BytesMessage) message;
                    try {
                        String msgStr = ActiveMQUtils.byteMessageToString(byteMessage);
                        System.out.println(ActiveMQUtils.formatMsg("接收消息：" + "BytesMessage", Thread.currentThread() + "-" + listenerName, msgStr));
                    } catch (JMSException e) {
                        e.printStackTrace();
                    }
                } else if (message instanceof MapMessage) {
                    MapMessage mapMessage = (MapMessage) message;
                    String msgStr = "";
                    System.out.println(ActiveMQUtils.formatMsg("接收消息：" + "MapMessage", Thread.currentThread() + "-" + listenerName, msgStr));
                } else if (message instanceof ObjectMessage) {
                    String msgStr = "";
                    System.out.println(ActiveMQUtils.formatMsg("接收消息：" + "ObjectMessage", Thread.currentThread() + "-" + listenerName, msgStr));
                } else if (message instanceof StreamMessage) {
                    String msgStr = "";
                    System.out.println(ActiveMQUtils.formatMsg("接收消息：" + "StreamMessage", Thread.currentThread() + "-" + listenerName, msgStr));
                } else {
                    System.out.println("接收消息：" + "消息类型：Unknown");
                }
            }
        });
    }

    public static void main(String[] args) {
        MqConnectionEntity connectionEntity = MqConnectionEntity.newBuilder()
                .setUrl("tcp://localhost:61616")
                .setUsername("admin")
                .setPassword("admin")
                .setDestination("queueTest")
                //.setDestinationType("queue")
                .setPubSubDomain(false)
                .setMessage("hello")
                .setByteMessage(true)
                .build();
        try {
            ActiveMQUtils.listener(connectionEntity, "listener1");
            System.out.println("listerer1");
            //ActiveMQUtils.pushToMq(connectionEntity);
            //System.out.println("publish1");
        } catch (JMSException e) {
            e.printStackTrace();
        }
    }
}
